package com.lagou.rpc.consumer.zookeeper;

import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.pojo.ServiceInfo;
import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import com.lagou.rpc.consumer.task.ScheduleTask;
import com.lagou.rpc.consumer.task.SimpleTask;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author boting.guo
 * @date 2021/11/14 11:23
 */

@Component
public class ZkHelper {

	private CuratorFramework client;

	private Map<String, RpcClient> rpcClientMap = new HashMap<>();

	public Map<String, ServiceInfo> getClientMap() {
		return clientMap;
	}

	private Map<String, ServiceInfo> clientMap = new HashMap<>();

	private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);

	public ZkHelper() {
		init();
	}

	private void init() {

		client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(1000, 3));
		client.start();

		watch();

		executorService.scheduleWithFixedDelay(new ScheduleTask(client, this), 5, 5, TimeUnit.SECONDS);
		new Thread(new SimpleTask(client, this)).start();
	}

	private void watch() {
		CuratorCache curatorCache = CuratorCache.builder(client, "/server-registry").build();

		CuratorCacheListener listener = CuratorCacheListener.builder()
				.forPathChildrenCache("/server-registry", client, new PathChildrenCacheListener() {
					@Override
					public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {

						if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
							ChildData childData = event.getData();
							String domain = childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1);
							String[] pieces = domain.split(":");
							if (pieces.length != 2) {
								return;
							}

							RpcClient rpcClient = new RpcClient(pieces[0], Integer.valueOf(pieces[1]));
							rpcClientMap.put(domain, rpcClient);
							System.out.println("新加入节点: " + domain);

							String value = new String(childData.getData());
							String[] valPieces = value.split(":");

							IUserService userService = (IUserService) RpcClientProxy.createProxy(IUserService.class, rpcClient);
							ServiceInfo serviceInfo = new ServiceInfo();
							serviceInfo.setLastTime(Long.valueOf(valPieces[0]));
							serviceInfo.setUsedTime(Long.valueOf(valPieces[1]));
							serviceInfo.setiUserService(userService);
							clientMap.put(domain, serviceInfo);

						} else if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
							ChildData childData = event.getData();
							String domain = childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1);
							System.out.println("节点断开:" + domain);
							rpcClientMap.remove(domain);
							clientMap.remove(domain);
						}
					}
				}).build();
		curatorCache.listenable().addListener(listener);
		curatorCache.start();
	}
}
